/*
 * (C) 2009, Elephant Seven Hamburg Gmbh, 
 * Author: Detlev Wiese
 * 
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not 
 * use this file except in compliance with the License. You may obtain a copy 
 * of the License at
 *  
 *     http://www.apache.org/licenses/LICENSE-2.0 
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.se and its accompanying notice file is
 * included with the distribution. 
 */ 
package com.e7.streams;

import java.util.ArrayList;
import java.util.List;

/**
 * <P>A stream class which allows reading elements again, which have 
 * already been delivered.</P> 
 * <P>To repeat reading elements, you just ask streams of this class 
 * for their "continued stream" using their method 
 * {@link ContinuedStream#continuedStream()}. The continued stream of a 
 * stream is simply the same stream but with a different state of 
 * elements delivered.</P>
 * <P>When you create an instance of a continued stream, all elements 
 * which are delivered by the base stream, will be recorded. N.B.: take 
 * memory consumption into account.</P> 
 * @param <E> The type of the elements
 */
public class ContinuedStream<E> extends AbstractStream<E> {

	private MemoryBackedStream<E> backingStream ; 
	private State state; 	
	
	/**
	 * Create an instance!
	 * @param stream The stream which should be wrapped to offer
	 * continued streams.  
	 */
	public ContinuedStream(Stream<E> stream) {
		this.backingStream = new MemoryBackedStream<E>(stream); 
		this.state = new State();
		backingStream.supportState(state);   
	}

	private ContinuedStream(MemoryBackedStream<E> stream, State state) {
		this.backingStream = stream; 
		this.state = state;
		backingStream.supportState(state);  
	}
	
	public boolean hasNext() {
		return backingStream.hasNext(state); 
	}

	public E next() {
		return backingStream.next(state); 
	}
	
	/**
	 * @return A new stream instance which is one of the continued streams 
	 * of the receiver. 
	 */
	public ContinuedStream<E> continuedStream() {
		backingStream.beRecording(); 
		return new ContinuedStream<E>(backingStream, new State(state)); 
	}
	
	/**
	 * @return True, if the receiver is in recording state, which means 
	 * it has delivered at least one continued stream.
	 */
	public boolean isRecording() { 
		return backingStream.isRecording; 
	}
	
	/**
	 * @return The number of elements recorded so far.
	 */
	public int recordedCount() {
		return backingStream.list.size(); 
	}

	/**
	 * <P>This method allows to compact the recordings of the receiver.</P> 
	 * <P>Once a continued stream is finalized by the Java virtual machine, 
	 * it will set its corresponding recording state as "not referenced". 
	 * That allows this method to later safely compact the elements which 
	 * still must be held available.</P> 
	 * <P>Use this method, if you must must keep memory consumption low.</P> 
	 */
	public void collectGarbage() { 
		backingStream.collectBackingGarbage();
	}
	
	protected void finalize() throws Throwable {
		state.setReferenced(false);
		state = null; 
		super.finalize();
	}
	
	private static class State {
		private int p; 
		private boolean referenced = true; 
		
		State() { p = 0; }
		
		State(State s) { p = s.position(); } 
		
		void advance() { p++; }
		
		int position() { return p; }

		public void setReferenced(boolean referenced) {
			this.referenced = referenced;
		}

		public boolean isReferenced() {
			return referenced;
		}
	}
	
	private static class MemoryBackedStream<E> {
		Stream<E> stream; 
		List<E> list = new ArrayList<E>(); 
		boolean isRecording; 
		List<State> supportedStates = new ArrayList<State>(); 
		
		MemoryBackedStream(Stream<E> stream) {
			this.stream = stream;
			isRecording = false; 
		}
		
		boolean hasNext(State state) {
			if (!isRecording) 
				return stream.hasNext(); 
			
			if (list.size() > state.position())
				return true; 
			else 
				return stream.hasNext(); 
		}
		
		E next(State state) {
			if (!isRecording) 
				return stream.next(); 
			
			E object; 
			if (list.size() > state.position())  
				object = list.get(state.position()); 
			else 
				list.add(object = stream.next());  
			
			state.advance(); 	
		
			return object; 
		}
		
		void supportState(State s) { 
			supportedStates.add(s); 
		}

		void beRecording() { 
			isRecording = true; 
		}
		
		synchronized void collectBackingGarbage() { 
			if (isRecording) { 
				int i = -1; 
				List<State> newStates = 
					new ArrayList<State>(supportedStates.size()); 
				
				for (State s : supportedStates) 
					if (s.isReferenced()) {
						int p = s.position(); 
						i = (-1 == i) ? p: Math.min(i, p); 
						newStates.add(s); 
					}
				if (i > 0) { 
					list = list.subList(i, list.size()); 
					for (State s : newStates) 
						s.p -= i; 
					
					supportedStates = newStates;
				}	
				// Recording to set off?
				if (supportedStates.isEmpty()) { 
					isRecording = false;
					list = new ArrayList<E>();
				}
			}
		}
	}
}
